package com.powerdata.core.paimon;

import com.powerdata.core.paimon.enums.PaimonBaseType;
import org.apache.commons.lang3.StringUtils;
import org.apache.paimon.predicate.In;
import org.apache.paimon.types.DataType;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.VarCharType;

import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

public class PDPaimonColumnUtils {
    List<String> types = Arrays.asList(
            "STRING",
            "FLOAT",
            "DOUBLE",
            "LONG",
            "INTEGER",
            "BOOLEAN",
            "DATE"
    );


    public static DataType transform(Map<String, String> entry) {
        String dataType = entry.get("dataType");

        switch (dataType) {
            case "VARCHAR":
                String length = entry.get("length");
                return DataTypes.VARCHAR(Integer.parseInt(length));
            case "STRING":
                return DataTypes.STRING();
            case "BOOLEAN":
                return DataTypes.BOOLEAN();
            case "FLOAT":
                return DataTypes.FLOAT();

            case "DOUBLE":
                return DataTypes.DOUBLE();

            case "LONG":
                return DataTypes.BIGINT();

            case "INTEGER":
                return DataTypes.INT();

            case "DATE":
                return DataTypes.DATE();
            case "Decimal":
                String precision = entry.get("precision");
                String scale = entry.get("scale");
                if (StringUtils.isBlank(precision) || StringUtils.isBlank(scale)) {
                    throw new RuntimeException("Decimal类型需要总位数和小数位！");
                }

                return DataTypes.DECIMAL(Integer.parseInt(precision), Integer.parseInt(scale));

            case "Map":


                String keyType = entry.get("keyType");
                PaimonBaseType keyTypeEnum = PaimonBaseType.getType(keyType);
                if (keyTypeEnum == null) {
                    throw new RuntimeException("map字段key未定义类型！");
                }
                String valueType = entry.get("valueType");
                PaimonBaseType valueTypeEnum = PaimonBaseType.getType(valueType);
                if (valueTypeEnum == null) {
                    throw new RuntimeException("map字段value未定义类型！");
                }
                return DataTypes.MAP(keyTypeEnum.getType(), valueTypeEnum.getType());

            case "Timestamp":
                if (StringUtils.equals(entry.get("isWithZone"), "1")) {
                    return DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE();
                } else {
                    return DataTypes.TIMESTAMP();
                }

            case "List":
                String listValueType = entry.get("valueType");
                PaimonBaseType listValueTypeEnum = PaimonBaseType.getType(listValueType);
                if (listValueTypeEnum == null) {
                    throw new RuntimeException("list字段元素未定义类型！");
                }
                return DataTypes.ARRAY(listValueTypeEnum.getType());
            default:
                break;
        }
        return null;
    }

    public static String transform(String columnType) {
        String upperCase = columnType.toUpperCase().replaceAll("\\(.*?\\)", "");
        List<String> collect = Arrays.stream(upperCase.split(" ")).collect(Collectors.toList());
        String columnTypeName = collect.get(0);
        switch (columnTypeName) {
            case "STRING":
                return "STRING";
            case "VARCHAR":
                return "STRING";
            case "DOUBLE":
                return "DOUBLE";
            case "BIGINT":
                return "LONG";
            case "INT":
                return "INTEGER";
            case "INTEGER":
                return "INTEGER";
            case "FLOAT":
                return "FLOAT";
            case "BOOLEAN":
                return "BOOLEAN";
            case "DATE":
                return "DATE";

        }
        //
        return null;
    }


    public static String realColumnName(String name) {
        List<String> realNames = Arrays.asList(name.replaceAll("[^(a-zA-Z0-9]", "").split(" ")).stream().collect(Collectors.toList());
        return realNames.get(0);
    }
}
